package mypack;

//import org.apache.hadoop.conf.Configuration;
//import org.apache.hadoop.fs.FSDataInputStream;
//import org.apache.hadoop.fs.FSDataOutputStream;
//import org.apache.hadoop.fs.FileSystem;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

@SuppressWarnings("deprecation")
public class Manager 
{	
	public static void CreateInitialPopulation() throws Exception
	{
		Job job= new Job();
	
		job.setJarByClass(MyDriver.class);
		//job.getConfiguration().setInt(Job.JVM_NUMTASKS_TORUN, -1);
		
		String inputPath; 
		String outputPath;
		
		inputPath = "input";
		outputPath =  "output-0";
		
		//inputPath =  "s3n://harun-bucket-2/MapReduceGen/input";
		//outputPath = "s3n://harun-bucket-2/MapReduceGen/output-0";
		
			
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
	
		job.setMapperClass(InitPopMapper.class);
	
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(Chromosome.class);
		
		job.setNumReduceTasks(0);
		
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
	
		job.waitForCompletion(true);
		
	}
	
	public static void evolveOneStep() throws Exception
	{	
		Job job= new Job();
		
		job.setJarByClass(MyDriver.class);
		job.setNumReduceTasks(Global.MAPREDUCE_NUMBER);
		//job.getConfiguration().setInt(Job.JVM_NUMTASKS_TORUN, -1);
		
		String inputPath; 
		String outputPath;

		inputPath = "output-"+Global.evolutionStep;
		//inputPath = "s3n://harun-bucket-2/MapReduceGen/output-"+Global.evolutionStep;
		System.out.println("input path: "+inputPath);
		
		Global.evolutionStep++;
		
		System.out.println("Step Number: "+Global.evolutionStep);
		
		outputPath =  "output-"+Global.evolutionStep;
		//outputPath = "s3n://harun-bucket-2/MapReduceGen/output-"+Global.evolutionStep;
		System.out.println("output path: "+outputPath);
			
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
	
		job.setMapperClass(EvolutionMapper.class);
		job.setPartitionerClass(PopulationPartitioner.class);
		job.setReducerClass(EvolutionReducer.class);
	
	
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(Chromosome.class);
		
		job.setInputFormatClass(SequenceFileInputFormat.class);
		job.setOutputFormatClass(SequenceFileOutputFormat.class);
	
		job.waitForCompletion(true);
	}
	
	public static void finishEvolution() throws Exception
	{
		Job job= new Job();
		
		job.setJarByClass(MyDriver.class);
		job.setNumReduceTasks(0);
		//job.getConfiguration().setInt(Job.JVM_NUMTASKS_TORUN, -1);
		
		String inputPath; 
		String outputPath;

		inputPath = "output-"+Global.evolutionStep;
		System.out.println("input path: "+inputPath);
		
		Global.evolutionStep++;
		System.out.println("Step Number: "+Global.evolutionStep);
		
		outputPath =  "output-"+Global.evolutionStep;
		System.out.println("output path: "+outputPath);
				
		FileInputFormat.addInputPath(job, new Path(inputPath));
		FileOutputFormat.setOutputPath(job, new Path(outputPath));
	
		job.setMapperClass(EvolutionMapper.class);
	
	
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Chromosome.class);
		
		job.setInputFormatClass(SequenceFileInputFormat.class);
	
		job.waitForCompletion(true);
	}
	
	public static boolean isGoodEnough(int generationNumber) throws IOException
	{
		FileSystem fs = FileSystem.get(new Configuration());
		int bestSolutionNeeded = Global.FITNESS_VAL_NEEDED;
		
		for (int i = 0; i < Global.MAPREDUCE_NUMBER; i++) 
		{
			String bestFileName = "result-"+generationNumber+"/bestInPopulation-"+(i+1)+".txt";
			Path inFile = new Path(bestFileName);
			if (!fs.exists(inFile))
			{
				 Population.writeLog("file_best", ""+generationNumber, "file not exist");
				 continue;
			}
			FSDataInputStream in = fs.open(inFile);
			String line = in.readLine().trim();
			int solution = Integer.parseInt(line);
			if (solution <= bestSolutionNeeded) 
			{
				return true;
			}	
			in.close();
		}
		
		return false;
	}
}
